package cn.monkey.transport.mq.redisson;

import cn.monkey.transport.core.Exchange;
import org.redisson.api.RedissonClient;

import java.util.Optional;

public class SimpleRedissonExchangeFactory extends AbstractRedissonExchangeFactory<RedissonExchange> {


    public SimpleRedissonExchangeFactory(RedissonClient redissonClient) {
        super(redissonClient);
    }

    public SimpleRedissonExchangeFactory(RedissonClient inboundClient, RedissonClient outBoundClient) {
        super(inboundClient, outBoundClient);
    }

    @Override
    protected RedissonInbound newInbound(String key, RedissonTransportConfig.Inbound inbound) {
        String inboundTopic = Optional.of(inbound)
                .map(RedissonTransportConfig.Inbound::getTopic)
                .filter(s -> !s.isEmpty())
                .orElse(key + "_inbound");
        return new SimpleRedissonInbound(inboundTopic, this.inboundClient, inbound.getCodec(),
                inbound.getPollSize(), inbound.getPollFrequencyMs());
    }

    @Override
    protected RedissonOutbound newOutbound(String key, RedissonTransportConfig.Outbound outbound) {
        String outboundTopic = Optional.of(outbound)
                .map(RedissonTransportConfig.Outbound::getTopic)
                .filter(s -> !s.isEmpty())
                .orElse(key + "_outbound");
        return new SimpleRedissonOutbound(outboundTopic, this.outBoundClient, outbound.getCodec(), outbound.getCacheSize());
    }

    @Override
    protected RedissonExchange inverse(RedissonExchange exchange) {
        return exchange.inverse();
    }

    @Override
    public RedissonExchange apply(String key, Exchange exchange, RedissonTransportConfig redissonExchangeConfig) {
        if (exchange == null) {
            SimpleRedissonExchange redissonExchange = new SimpleRedissonExchange(key, this.newInbound(key, redissonExchangeConfig.getInbound()), this.newOutbound(key, redissonExchangeConfig.getOutbound()));
            return redissonExchangeConfig.isInverse() ? super.inverse(redissonExchange).bind() : redissonExchange.bind();
        }
        if (exchange instanceof RedissonExchange redissonExchange) {
            return redissonExchange;
        }
        throw new IllegalArgumentException("exchange should be redissonExchange or null");
    }
}
